package com.junlinpro.utils;


import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaProducerUtil {
	
	private static KafkaProducer<String, byte[]> producer = null;
	private static String topic = "soundtopic";
	private static String realTimeTopic = "soundtopiclistening";
	
	static{
		Properties properties = new Properties();
		
		//1、指定kafkabroker地址
		properties.setProperty("bootstrap.servers", "10.1.52.3:9092,10.1.52.4:9092");

		//2、指定kv的序列化类
		properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
		properties.setProperty("max.request.size", "50000000");

		//创建生产者
		producer = new KafkaProducer<>(properties);    
	}
	
	public static void saveSound(String cardKey,byte[] data) {
		ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic,cardKey,data);
    	producer.send(record);        
    	producer.flush();
	}
	
	public static void saveRealTimeSound(String cardKey,byte[] data) {
		ProducerRecord<String, byte[]> record = new ProducerRecord<>(realTimeTopic,cardKey,data);
    	producer.send(record);        
    	producer.flush();
	}

}
